package com.rem.flink;

import cn.hutool.json.JSONUtil;
import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Set;

/**
 * @author Rem
 * @date 2022-09-30
 */

public class KafkaToRedis {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("10.224.18.157:9092,10.224.18.158:9092,10.224.18.159:9092")
                .setTopics("clicks")
                .setGroupId("consumer-group")
                .setStartingOffsets(OffsetsInitializer.committedOffsets())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        //   DataStreamSource<String> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

//        SingleOutputStreamOperator<Event> eventSource = kafkaSource.map(line -> {
//            String[] fields = line.split(" ");
//            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
//        });

        DataStreamSource<Event> eventSource = env.addSource(new ClickSource());

        SingleOutputStreamOperator<Event> count = eventSource.assignTimestampsAndWatermarks(
                        //插入水位线 针对乱序设置延迟时间  顺序不需要设置
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                //乱序设置2s的延迟时间
                                //forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                //提取时间戳
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()))
                .keyBy(Event::getUser)
                //设置滚动事件窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("count");
        count.print();
        env.execute();


        FlinkJedisSentinelConfig conf = new FlinkJedisSentinelConfig.Builder()
                //   .setSentinels(Set.of("10.224.18.157:26379", "10.224.18.158:26379", "10.224.18.159:26379"))
                .setSentinels(Set.of("localhost:6380", "localhost:6381", "localhost:6382"))
//                .setMasterName("RedisMaster")
                .setMasterName("mymaster")
//                .setPassword("Aldi2020!@#")
                .setPassword("123456")
                .build();

        eventSource.print();
        // eventSource.addSink(new RedisSink<Event>(conf, new MyRedisMapper()));


    }

    public static class MyRedisMapper implements RedisMapper<Event> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "clicks");
        }

        @Override
        public String getKeyFromData(Event event) {
            return event.getUser();
        }

        @Override
        public String getValueFromData(Event event) {
            return JSONUtil.toJsonStr(event);
        }
    }
}
